package com.huawei.dli.flink.sye.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

public class DCSSink extends RichSinkFunction<String> {
  private static final Logger logger = LoggerFactory.getLogger(DCSSink.class);

  private String dcsAddr;
  private String password;

  private transient Jedis jedisClient;

  private int index = 0;

  public DCSSink(String dcsAddr, String password) {
    this.dcsAddr = dcsAddr;
    this.password = password;
  }

  @Override
  public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    connectRedis();
  }

  @Override
  public void invoke(String in) {
    try {
      // 根据业务场景，调用jedis接口操作
      String result = jedisClient.set("key" + index, in);
      logger.info("Send data: [" + in + "] to dcs: " + result);
      index++;
    } catch (Exception ex) {
      logger.error("Failed to write data to dcs", ex);
    }
  }

  @Override
  public void close() throws Exception {
    super.close();

    if (jedisClient != null) {
      jedisClient.close();
    }
  }

  private void connectRedis() {
    logger.info("try connect ${config.ip}:${config.port}");
    String[] ipPort = dcsAddr.trim().split(":");
    jedisClient = new Jedis(ipPort[0], Integer.parseInt(ipPort[1]));
    if (password != null) {
      String authResponse = jedisClient.auth(password);
      if (!authResponse.equals("OK")) {
        String errMsg = "AUTH FAILED: wrong password";
        logger.error(errMsg);
        throw new SecurityException(errMsg);
      }
    }
  }
}
